package com.errol.sync.sync.wrapper;

import com.renjia.supervise.service.log.LogService;
import com.renjia.supervise.service.sync.DataSyncService;
import com.renjia.supervise.service.sync.pojo.EntityInfo;
import lombok.ToString;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.lang.NonNull;

import java.time.LocalDate;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 并发控制的 wrapper
 *
 * @author zhang xiao bin
 * @version v1.0
 * @date 2021/08/16
 * @since v
 **/
public class SmartDataSyncWrapper extends AbsDataSyncWrapper implements ApplicationListener<SmartDataSyncWrapper.SmartEvent> {

    /**
     * 构造时传入，由 {@link DataSyncWrapperFactory} 维护
     */
    private final ConcurrentMap<Class<?>, BlockingQueue<SyncTask>> syncMissionCache;

    /**
     * 用于创建 ApplicationListener 实例 {@link DataSyncWrapperFactory#smartDataSyncApplicationListener()}
     */
    SmartDataSyncWrapper(ConcurrentMap<Class<?>, BlockingQueue<SyncTask>> syncMissionCache, LogService logService) {
        super(null, logService);
        this.syncMissionCache = syncMissionCache;
    }

    /**
     * 迭代获取最里层的 source，替换为 SmartEventPublisher（发布任务完成或阻塞的事件）
     */
    SmartDataSyncWrapper(DataSyncService source, ApplicationContext applicationContext, ConcurrentMap<Class<?>, BlockingQueue<SyncTask>> syncMissionCache, LogService logService) {
        super(source, logService);
        this.syncMissionCache = syncMissionCache;
        AbsDataSyncWrapper w = this;
        DataSyncService s = this.source;
        while (s instanceof AbsDataSyncWrapper) {
            w = (AbsDataSyncWrapper) s;
            s = ((AbsDataSyncWrapper) s).source;
        }
        w.source = new SmartEventPublisher(s, logService, applicationContext);
    }

    /**
     * 每个实体类都有一个阻塞队列，将同步任务封装为 SyncTask 加入阻塞队列：
     * 1. 如果是队列第一个元素，触发同步任务，否则等待执行中的同步任务完成（通过事件回调）
     * 2. 任务等待时，检查是否有状态异常的任务，及进行任务合并（待实现）
     */
    @Override
    public <E> String syncData(Class<E> entityClass, LocalDate rangeLeft, LocalDate rangeRight, boolean updateEditedData) {
        BlockingQueue<SyncTask> blockingQueue = syncMissionCache.computeIfAbsent(entityClass, ec -> new LinkedBlockingQueue<>(32));
        SyncTask syncTask = new SyncTask(entityClass, rangeLeft, rangeRight, updateEditedData, source);
        if (!blockingQueue.offer(syncTask)) {
            return refusedWarningMsg(entityClass);
        } else if (syncTask == blockingQueue.peek()) {
            // TODO: 21/8/20 如果链路中的某个 wrapper 拦截了任务，或者合并了任务，将导致该队列阻塞
            // 第一个任务执行同步，后续任务进入等待队列，待收到第一个任务的结束回调，触发第二个任务
            return syncTask.source.syncData(entityClass, rangeLeft, rangeRight, updateEditedData);
        } else {
            // TODO: 21/8/20 分析合并任务
            long threshold = System.currentTimeMillis() - 10800000;
            // 队列首元素加入队列已超过3小时，执行容错处理
            if (blockingQueue.element().startTime < threshold) {
                EntityInfo<E> entityInfo = source.getEntityInfo(entityClass);
                // 队列过长、或首元素加入队列已超过2小时，并且能对 entityClass 加锁成功，则同步队列中的任务状态可能存在异常
                if (entityInfo != null && entityInfo.getInSync().compareAndSet(false, true)) {
                    synchronized (entityClass) {
                        SyncTask first = blockingQueue.peek();
                        while (first != null && first.startTime < threshold) {
                            if (!blockingQueue.remove(first)) {
                                // 回调执行中，返回阻塞提示
                                return blockedWarningMsg(entityClass);
                            }
                            first = blockingQueue.peek();
                        }
                        entityInfo.getInSync().compareAndSet(true, false);
                        logService.remind(entityClass.getSimpleName() + "同步队列触发容错处理");
                        if (first != null) {
                            return first.source.syncData(first.entityClass, first.rangeLeft, first.rangeRight, first.updateEditedData);
                        }
                    }
                }
            }
            return blockedWarningMsg(entityClass);
        }
    }

    /**
     * {@link SmartEventPublisher} 发布的阻塞或完成的事件
     * 注意：如果 source 链路中的 wrapper 拦截的任务，则无法触发该事件回调
     */
    @Override
    public void onApplicationEvent(@NonNull SmartEvent event) {
        // 使用 @Async 线程池，可能被占满而无法即时通知到，有可能因执行顺序导致通知乱序到达
        Class<?> entityClass = (Class<?>) event.getSource();
        try {
            BlockingQueue<SyncTask> blockingQueue = syncMissionCache.get(entityClass);
            SyncTask first = blockingQueue.element();
            switch (event.eventType) {
                case BLOCKED:
                    // smart wrapper 以外，有其他线程正在同步表，睡眠一分钟后再次尝试同步
                    first.startTime = event.startTime;
                    try {
                        Thread.sleep(60000);
                    } catch (InterruptedException e) {
                        // never mind
                    }
                    first.source.syncData(entityClass, first.rangeLeft, first.rangeRight, first.updateEditedData);
                    return;
                case FINISHED:
                    first.startTime = event.startTime;
                    first.finishTime = event.finishTime;
                    first.errMsg = event.errMsg;
                    if (event.errMsg != null) {
                        try {
                            // 同步失败，等待10分钟后开始下个任务
                            logService.warn("同步任务执行失败：" + first.toString());
                            logService.remind(entityClass.getName() + " 同步任务执行失败：" + first.errMsg);
                            Thread.sleep(600000);
                        } catch (InterruptedException e) {
                            // never mind
                        }
                    }
                    blockingQueue.remove(first);
                    first = blockingQueue.peek();
                    if (first != null) {
                        first.source.syncData(entityClass, first.rangeLeft, first.rangeRight, first.updateEditedData);
                    }
                    return;
                default:
                    throw new Exception("未处理的事件类型");
            }
        } catch (Exception e) {
            logService.error(e);
            logService.remind(entityClass.getName() + " 同步任务回调异常：" + e.getMessage());
            // 理论上只有回调触发的 syncData 可能异常，此时会有新的回调通知，所以这里输出异常就行
        }
    }

    /**
     * 同步任务对象
     */
    @ToString
    static class SyncTask {
        final Class<?> entityClass;
        final LocalDate rangeLeft;
        final LocalDate rangeRight;
        final boolean updateEditedData;
        final DataSyncService source;

        final long addTime;
        long startTime;
        long finishTime;
        String errMsg;

        public SyncTask(Class<?> entityClass, LocalDate rangeLeft, LocalDate rangeRight, boolean updateEditedData, DataSyncService dataSyncService) {
            this.entityClass = entityClass;
            this.rangeLeft = rangeLeft;
            this.rangeRight = rangeRight;
            this.updateEditedData = updateEditedData;
            this.addTime = System.currentTimeMillis();
            this.startTime = this.addTime;
            this.source = dataSyncService;
        }

        public boolean isFinished() {
            return startTime > 0 && finishTime > 0;
        }
    }

    /**
     * 同步任务事件发布器，source 是同步服务实例
     * 执行完成或加锁失败发布事件通知 SmartDataSyncWrapper
     */
    static class SmartEventPublisher extends AbsDataSyncWrapper {
        private final ApplicationContext applicationContext;

        public SmartEventPublisher(DataSyncService source, LogService logService, ApplicationContext applicationContext) {
            super(source, logService);
            this.applicationContext = applicationContext;
        }

        @Override
        public <E> String syncData(Class<E> entityClass, LocalDate rangeLeft, LocalDate rangeRight, boolean updateEditedData) {
            EntityInfo<E> entityInfo = source.getEntityInfo(entityClass);
            long startTime = System.currentTimeMillis();
            if (entityInfo == null) {
                String npeMsg = source.syncData(entityClass, rangeLeft, rangeRight, updateEditedData);
                applicationContext.publishEvent(new SmartEvent(entityClass, SmartEvent.EventType.FINISHED, npeMsg, startTime, System.currentTimeMillis()));
                return npeMsg;
            } else {
                if (entityInfo.getInSync().compareAndSet(false, true)) {
                    String errMsg = source.syncData(entityClass, rangeLeft, rangeRight, updateEditedData);
                    applicationContext.publishEvent(new SmartEvent(entityClass, SmartEvent.EventType.FINISHED, errMsg, startTime, System.currentTimeMillis()));
                    return errMsg;
                } else {
                    applicationContext.publishEvent(new SmartEvent(entityClass, SmartEvent.EventType.BLOCKED, null, startTime, 0));
                    return blockedWarningMsg(entityClass);
                }
            }
        }
    }

    /**
     * 同步任务事件
     */
    static class SmartEvent extends ApplicationEvent {

        final EventType eventType;
        final String errMsg;
        final long startTime;
        final long finishTime;

        public SmartEvent(Object source, EventType eventType, String errMsg, long startTime, long finishTime) {
            super(source);
            this.eventType = eventType;
            this.errMsg = errMsg;
            this.startTime = startTime;
            this.finishTime = finishTime;
        }

        enum EventType {
            BLOCKED,
            FINISHED,
        }
    }

    static String blockedWarningMsg(Class<?> entityClass) {
        return entityClass.getSimpleName() + "已加入同步队列，等待执行";
    }

    static String refusedWarningMsg(Class<?> entityClass) {
        return entityClass.getSimpleName() + "同步队列已满，请稍后再试";
    }

}
